package com.ruozedata.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.net.URI;

/**
 * 1) 统计指定目录下[传几个大于blocksize的文件]，不足blocksize数据块的占比
 * 	1file ==> 3个block  1/3
 * 2) /xxxx/
 * 		a.java
 * 		b.java
 * 		c.jpg
 * 		d.scala
 * 	统计指定目录下某个后缀名的文件个数，并删除
 * 		xxxx(String suffix) {  // scala }
 *
 */
public class HDFSApp {

    private static String path = "/tmp";

    public static void main(String[] args) throws Exception {
        // 统计不足blocksize数据块的占比
        getNotFullyUsedBlockPct(path);

        // 统计指定目录下某个后缀名文件个数，并删除
        countFile(path, "scala");
    }


    // 获取FileSystem对象
    public static FileSystem getFileSystem() throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("dfs.client.use.datanode.hostname", "true");
        URI uri = new URI("hdfs://hadoop000:8020");
        return FileSystem.get(uri, configuration, "hadoop");
    }

    /**
     * 统计指定目录下未完全使用的file block占比
     * @param path  统计目录，空则为根目录
     */
    public static void getNotFullyUsedBlockPct(String path) throws Exception {
        int totalBlockNum = 0;
        int unFullUsedBlockNum = 0;
        int filesCount = 0;
        FileSystem fileSystem = getFileSystem();
        // 获取所有文件
        RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path(path), true);
        // 循环获取所有文件及文件夹信息
        while (files.hasNext()) {
            LocatedFileStatus file = files.next();
            int blockCount = file.getBlockLocations().length;
            filesCount += 1;
            totalBlockNum += blockCount;
            System.out.println("file: "+ file.getPath().toString() + "-------------------------------------");
            System.out.println("该文件共有 [\" + blockCount + \"] 个block, ");
            if (!file.isDirectory()) {
                BlockLocation[] blockLocations = file.getBlockLocations();
                for (int i = 0; i < blockLocations.length; i++) {
                    long blockSize = blockLocations[i].getLength();
                    if (blockSize < 128 * 1024 * 1024) unFullUsedBlockNum += 1;
                    System.out.println("" + "block[" + i + "] 大小为：[" + blockSize + "Bytes]");
                }
            }
        }

        double res = totalBlockNum == 0 ? 0 : unFullUsedBlockNum / (double) totalBlockNum;
        System.out.println(path + "目录下，文件数[" + filesCount + "]，block数[" + totalBlockNum + "], 空间未完全使用block数[" + unFullUsedBlockNum + "], 占比: " + res);
    }


    /**
     * 统计指定目录下指定文件后缀名，文件的个数
     * @param path      指定目录
     * @param suffix    指定文件后缀名（不包含.）
     */
    public static void countFile(String path, String suffix) throws Exception {
        int count = 0;

        FileSystem fileSystem = getFileSystem();
        RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path(path), true);
        while (files.hasNext()) {
            LocatedFileStatus file = files.next();
            String fileName = file.getPath().getName();
            String fileSuffix = fileName.contains(".") ? fileName.split("\\.")[fileName.split("\\.").length - 1] : "";
            if (suffix.equalsIgnoreCase(fileSuffix)) count++;

            // 删除该文件
            fileSystem.delete(file.getPath(), true);
        }
        System.out.println(path + "目录下，" + suffix + " 后缀名的文件个数为[" + count + "]个");
    }
}
